﻿//========================================================= 
/**@file CommandAsyncWorker.h
 * @brief Command异步任务执行器
 * 
 * @date 2017-07-05   18:25:30
 * @author Zhyioo 
 * @version 1.0
 */ 
//--------------------------------------------------------- 
#ifndef _LIBZHOUYB_COMMANDASYNCWORKER_H_
#define _LIBZHOUYB_COMMANDASYNCWORKER_H_
//--------------------------------------------------------- 
#ifdef INCLUDE_BOOST_SOURCE
/**
 * @brief 如果需要以嵌入源码的形式使用async_namespace::THREAD库,则在外部定义下列宏,并包含CommandWorker.cpp文件
 * @date 2017-06-29 00:20:30
 */
/*
#define BOOST_ALL_NO_LIB
#define BOOST_DATE_TIME_SOURCE
#define BOOST_THREAD_NO_LIB
*/

#include <boost/shared_ptr.hpp>
#include <boost/make_shared.hpp>
#include <boost/thread/thread.hpp>

#define async_namespace boost
//--------------------------------------------------------- 
#else
//--------------------------------------------------------- 
#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <vector>

#define async_namespace std
//--------------------------------------------------------- 
#endif
//--------------------------------------------------------- 
#include "CommandWorker.h"
//--------------------------------------------------------- 
namespace zhou_yb {
namespace application {
namespace driver {
namespace async {
//--------------------------------------------------------- 
/**
 * @brief 命令任务数据结构
 * @date 2017-06-23 18:08
 */
struct TaskDescription
{
    /// 需要执行的命令
    Ref<ICommandHandler> WorkCmd;
    /// 用于回调的命令
    Ref<ICommandHandler> CallbackCmd;
    /// 执行任务的参数
    StringArguments Argument;
    /// 执行任务的结果
    StringArguments Result;
    /// 是否为后台任务
    bool IsBackground;
    /// 上次调度时间
    uint TaskTime;
    /// 任务的最小唤醒超时时间
    uint WakeUpTimeout;
};
//--------------------------------------------------------- 
/**
 * @brief 命令任务执行器
 * @date 2017-06-23 17:32
 *
 * 维护一个命令构成的消息队列,依次执行队列中的命令
 */
class CommandTaskExecutor : public RefObject
{
    DisableObjConsturctCopyBehavior(CommandTaskExecutor)
public:
    typedef const TaskDescription* const RefTask;
    /// 超时时间最大值
    static const uint TIMEOUT_EOF = static_cast<uint>(-1);
    /// 返回是否为任务参数
    inline static bool IsTaskProperty(const char* name)
    {
        return StringConvert::StartWith(name, "Task::");
    }
    /// 返回任务描述符
    inline static RefTask Task(IStringArguments& arg)
    {
        const char TASK_REFTASK[] = "Task::RefTASK";
        pointer pTask = arg[TASK_REFTASK].To<pointer>(NULL);
        if(pTask != NULL)
            return reinterpret_cast<TaskDescription*>(pTask);
        return NULL;
    }
    /// 返回任务是否执行成功
    inline static bool IsTaskOK(IStringArguments& arg)
    {
        const char TASK_ISOK[] = "Task::IsOK";
        return arg[TASK_ISOK].To<bool>(false);
    }
    /// 设置是否为后台任务
    inline static void SetIsBackground(IStringArguments& arg, bool isBackground)
    {
        TaskDescription* task = const_cast<TaskDescription*>(Task(arg));
        if(task != NULL)
        {
            task->IsBackground = isBackground;
        }
    }
    /// 设置任务为后台带唤醒功能的任务
    inline static void SetWakeUpTask(TaskDescription& task, uint wakeupTime)
    {
        task.IsBackground = true;
        task.WakeUpTimeout = wakeupTime;
        task.TaskTime = Timer::TimeTick();
    }
protected:
    list_t<TaskDescription> _queue;
    list_t<TaskDescription> _cached_queue;
    async_namespace::condition_variable_any _condition;
    async_namespace::shared_ptr<async_namespace::thread> _workThread;
    async_namespace::mutex _mutex;
    BoolInterrupter _interrupter;
    Ref<ICommandHandler> _taskCancelCmd;

    /// 用于在任务线程中取消的命令
    LC_CMD_METHOD(TaskCancel)
    {
        _interrupter.Interrupt();
        return true;
    }
    /// 执行任务
    void Run()
    {
        const char TASK_ISOK[] = "Task::IsOK";
        const char TASK_REFTASK[] = "Task::RefTASK";

        list_t<TaskDescription>::iterator workItr;
        uint wakeupTime = TIMEOUT_EOF;
        size_t taskIndex = 0;
        while(true)
        {
            if(InterruptBehavior::Implement(_interrupter))
                break;

            {
                async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
                // 检查是否有任务需要执行
                if(_queue.size() > 0)
                {
                    ++taskIndex;
                    workItr = _queue.snip(_queue.begin());
                }
                else
                {
                    _condition.wait(lock);
                    continue;
                }
            }
            // 检测是否到唤醒时间
            if(workItr->IsBackground && workItr->WakeUpTimeout > 0)
            {
                uint currentTick = Timer::TimeTick();
                if(currentTick < workItr->TaskTime)
                {
                    async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
                    wakeupTime = _min(wakeupTime, (workItr->TaskTime - currentTick));
                    _queue.splice(workItr);
                    continue;
                }
                wakeupTime = _min(wakeupTime, workItr->WakeUpTimeout);
                workItr->TaskTime = (currentTick + workItr->WakeUpTimeout);
            }
            bool isWorkNull = workItr->WorkCmd.IsNull();
            bool bOK = false;
            // 如果WorkCmd为空则对象可能已经析构,仍然触发一次回调
            if(!isWorkNull)
            {
                // 运行Work命令时,保证隔离性,不添加 Task 相关属性参数到Argument中
                bOK = workItr->WorkCmd->OnCommand(workItr->Argument, workItr->Result);
            }
            if(!workItr->CallbackCmd.IsNull())
            {
                workItr->Argument.Put(TASK_ISOK, ArgConvert::ToString(bOK));
                workItr->Argument.Put(TASK_REFTASK, ArgConvert::ToString<pointer>(&(*workItr)));

                workItr->CallbackCmd->OnCommand(workItr->Argument, workItr->Result);
            }
            // 后台任务
            if(workItr->IsBackground && !isWorkNull)
            {
                async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
                workItr->Argument.Remove(TASK_ISOK);
                workItr->Argument.Remove(TASK_REFTASK);
                workItr->Result.Clear();
                _queue.splice(workItr);

                // 已经执行完一遍任务队列
                if(taskIndex >= _queue.size())
                {
                    taskIndex = 0;
                    if(wakeupTime > 0)
                    {
                        #ifdef INCLUDE_BOOST_SOURCE
                        _condition.timed_wait(lock, async_namespace::posix_time::milliseconds(wakeupTime));
                        #else
                        _condition.wait_for(lock, std::chrono::milliseconds(wakeupTime));
                        #endif
                    }
                    wakeupTime = TIMEOUT_EOF;
                }
            }
            else
            {
                async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
                _queue.splice(workItr);
                _queue.pop_back();
            }
        }
    }
    /// 设置任务属性
    TaskDescription& GetTask(list_t<TaskDescription>& queue, Ref<ICommandHandler> work, 
        Ref<ICommandHandler> callback, bool isBackground, uint wakeupTime)
    {
        list_t<TaskDescription>::iterator itr = queue.push_back();
        itr->WorkCmd = work;
        itr->CallbackCmd = callback;
        itr->IsBackground = isBackground;
        if(isBackground)
        {
            itr->WakeUpTimeout = wakeupTime;
            // 设置后立即执行
            itr->TaskTime = Timer::TimeTick();
        }
        else
        {
            itr->TaskTime = 0;
            itr->WakeUpTimeout = 0;
        }
        return (*itr);
    }
    /// 启动任务
    bool DoTask()
    {
        // 如果该命令为第一个命令,则开启工作线程
        if(_workThread == NULL)
        {
            _interrupter.Reset();
            _workThread = async_namespace::make_shared<async_namespace::thread>(
                async_namespace::bind(&CommandTaskExecutor::Run, this));
        }
        _condition.notify_one();
        return true;
    }
public:
    CommandTaskExecutor()
    {
        _workThread = NULL;
    }
    virtual ~CommandTaskExecutor()
    {
        Cancel();
    }
    /**
     * @brief 返回任务队列是否在执行
     * @date 2019-03-17 18:31:45
     */
    inline bool IsRunning() const
    {
        return _workThread != NULL;
    }
    /**
     * @brief 返回用于在callback中取消任务的命令
     * @date 2019-03-17 19:01:20
     */
    Ref<ICommandHandler> CancelCommand()
    {
        if(_taskCancelCmd.IsNull())
        {
            _taskCancelCmd = _MakeCMD(CommandTaskExecutor, TaskCancel);
        }
        return _taskCancelCmd;
    }
    /**
     * @brief 准备任务,返回任务的参数引用
     * @date 2019-03-17 11:40:23
     */
    IStringArguments& Prepare(Ref<ICommandHandler> work, Ref<ICommandHandler> callback = Ref<ICommandHandler>(),
        bool isBackground = false, uint wakeupTimeMs = 0)
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        TaskDescription& task = GetTask(_cached_queue, work, callback, isBackground, wakeupTimeMs);
        return task.Argument;
    }
    /**
     * @brief 清空已准备好的任务
     * @date 2019-03-14 23:02:49
     */
    size_t Revise()
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        size_t count = _cached_queue.size();
        _cached_queue.clear();
        return count;
    }
    /**
     * @brief 完成准备任务,将所有已准备好的任务放到队列中执行
     * @date 2019-03-14 22:59:49
     */
    size_t Done()
    {
        size_t count = _cached_queue.size();
        if(count > 0)
        {
            async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
            if(_queue.empty())
            {
                _queue.swap(_cached_queue);
            }
            else
            {
                _queue.splice(_queue.end() - 1, _cached_queue);
            }
            DoTask();
        }
        return count;
    }
    /**
     * @brief 新建任务,并立即添加到队列中依次执行
     * @date 2017-06-29 10:48
     *
     * @param [in] work 需要执行的任务
     * @param [in] pArg [default:Null] 执行命令时的参数
     * @param [in] callback [default:Null] 回调
     * - 回调参数: arg
     *  - WorkCmd : string 任务名称
     *  - IsOK : bool 任务执行结果
     *  - IsBackground : bool 是否为后台任务(可在回调中取消)
     *  - RefTASK : pointer TaskDescription指针(在回调函数生命周期内有效)
     * .
     * - 回调结果: rlt
     *  - 命令任务的输出结果 rlt
     * .
     * @param [in] isBackground [default:false] 是否为后台任务
     * @param [in] wakeupTime [default:0] 为后台任务时的唤醒时间间隔
     */
    bool OnTask(Ref<ICommandHandler> work, const IStringArguments* pArg = NULL,
        Ref<ICommandHandler> callback = Ref<ICommandHandler>(), 
        bool isBackground = false, uint wakeupTime = 0)
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        TaskDescription& task = GetTask(_queue, work, callback, isBackground, wakeupTime);
        if(pArg != NULL)
        {
            StringArguments::CopyTo(*pArg, task.Argument);
        }
        return DoTask();
    }
    #ifdef INCLUDE_BOOST_SOURCE
    /**
     * @brief 等待所有任务
     * @date 2017-06-29 10:07
     *
     * @param [in] timeoutMs 等待的最大超时时间,为0则表示一直等待直到完成
     */
    bool Join(uint timeoutMs = 0)
    {
        if(!IsRunning())
            return false;
        bool bWait = true;
        if(timeoutMs > 0)
        {
            bWait = _workThread->timed_join(async_namespace::posix_time::milliseconds(timeoutMs));
        }
        else
        {
            _workThread->join();
        }
        return bWait;
    }
    #else
    /// 等待所有任务执行完
    bool Join()
    {
        if(!IsRunning())
            return false;
        _workThread->join();
        _workThread = NULL;
        return true;
    }
    #endif
    /// 在回调中取消执行任务
    void OnCancel()
    {
        _interrupter.Interrupt();
    }
    /// 在回调中清空任务队列(包括缓存的任务)
    void OnClear()
    {
        _queue.clear();
        _cached_queue.clear();
    }
    /**
     * @brief 等待当前队列中的前台任务完成
     * @date 2019-03-17 18:30:26
     */
    void Wait()
    {
        if(!IsRunning())
            return;
        OnTask(CancelCommand());
        _workThread->join();
        _workThread = NULL;
    }
    /**
     * @brief 取消所有任务
     * @date 2017-06-29 10:47
     *
     * @warning 在callback回调中进行取消时,不能直接调用Cancel接口,会抛出异常,使用CancelCommand作为回调或者调用OnCancel
     *
     * @param [in] timeoutMs 等待中断的最大延时时间
     */
    void Cancel(uint timeoutMs = DEV_WAIT_TIMEOUT)
    {
        if(!IsRunning())
            return;

        InterruptBehavior::Interrupt(_interrupter, timeoutMs);
        _condition.notify_one();

        #ifdef INCLUDE_BOOST_SOURCE
        _workThread->interrupt();
        #endif
        _workThread->join();
        _workThread = NULL;
    }
    /**
     * @brief 返回当前任务数
     * @date 2017-06-29 10:48
     */
    size_t Count()
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        size_t count = _queue.size();
        return count;
    }
    /**
     * @brief 清空任务队列
     * @date 2017-06-29 10:48
     */
    void Clear()
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        OnClear();
    }
};
//--------------------------------------------------------- 
/**
 * @brief 基于CommandSetter的命令调度器
 *
 * 使用TryAsync时,前台任务只能有一个,可有多个后台任务并行
 *
 * @date 2017-06-24 11:18
 */
class CommandAsyncDispatcher : public CommandTaskExecutor
{
    DisableObjConsturctCopyBehavior(CommandAsyncDispatcher)
protected:
    string _currentCmd;
    async_namespace::mutex _tryMutex;
    Ref<ICommandHandler> _callback;
    Ref<ICommandHandler> _callbackCmd;
    LC_CMD_METHOD(OnCallback)
    {
        if(!_callback.IsNull())
        {
            _callback->OnCommand(arg, rlt);
        }
        _callback.Free();
        _tryMutex.unlock();
        return true;
    }
    void PrepareCallback(Ref<ICommandHandler> callback)
    {
        if(_callbackCmd.IsNull())
        {
            _callbackCmd = _MakeCMD(CommandAsyncDispatcher, OnCallback);
        }
        _callback = callback;
    }
    DeviceError::ErrorCode OnAsync(Ref<ICommandHandler> complexCmd, const IStringArguments* pArg, 
        Ref<ICommandHandler> callback, bool isBackground, uint wakeupTimeMs)
    {
        bool bOK = CommandTaskExecutor::OnTask(complexCmd, pArg, callback, isBackground, wakeupTimeMs);
        return bOK ? DeviceError::Success : DeviceError::OperatorStatusErr;
    }
public:
    CommandAsyncDispatcher() : CommandTaskExecutor() {}
    LC_CMD_METHOD(TryLock)
    {
        return _tryMutex.try_lock();
    }
    LC_CMD_METHOD(Lock)
    {
        _tryMutex.lock();
        return true;
    }
    LC_CMD_METHOD(Unlock)
    {
        _tryMutex.unlock();
        return true;
    }
    /**
     * @brief 尝试执行有互斥性的命令(例如上下电互斥,但是和卡片状态不互斥可并行)
     * @date 2017-06-29 22:19:40
     *
     * @param [in] cmd 需要执行的命令
     * @param [in] pArg 执行命令时的参数
     * @param [in] callback 调用命令后的回调函数
     *
     * @return DeviceError::ErrorCode 错误码
     * - 错误码:
     *  - DeviceError::DevIsBusyErr 已经有同类命令正在执行
     *  - DeviceError::ArgIsNullErr 指定名称的命令不存在
     *  - DeviceError::Success 调度成功
     * .
     */
    DeviceError::ErrorCode TryAsync(Ref<ICommandHandler> cmd, const IStringArguments* pArg = NULL, 
        Ref<ICommandHandler> callback = Ref<ICommandHandler>())
    {
        if(!_tryMutex.try_lock())
            return DeviceError::DevIsBusyErr;
        PrepareCallback(callback);
        return OnAsync(cmd, pArg, _callbackCmd, false, 0);
    }
    /**
     * @brief 添加任务到队列中
     * @date 2017-06-29 16:39
     *
     * @param [in] cmd 需要添加的命令
     * @param [in] pArg 调度命令时的参数
     * @param [in] callback 调用命令后的回调函数
     *
     * @return DeviceError::ErrorCode 错误码
     * - 错误码:
     *  - DeviceError::ArgIsNullErr 指定名称的命令不存在
     *  - DeviceError::Success 调度成功
     * .
     */
    DeviceError::ErrorCode Async(Ref<ICommandHandler> cmd, const IStringArguments* pArg = NULL, 
        Ref<ICommandHandler> callback = Ref<ICommandHandler>())
    {
        return OnAsync(cmd, pArg, callback, false, 0);
    }
    /**
     * @brief 添加后台任务到队列中
     * @date 2017-08-09 21:27
     * 
     * @param [in] cmd 需要添加的命令
     * @param [in] pArg 调度命令时的参数
     * @param [in] callback 调用命令后的回调函数
     * 
     * @return DeviceError::ErrorCode 错误码
     * - 错误码:
     *  - DeviceError::ArgIsNullErr 指定名称的命令不存在
     *  - DeviceError::Success 调度成功
     * .
     */
    DeviceError::ErrorCode Background(Ref<ICommandHandler> cmd, const IStringArguments* pArg = NULL, 
        Ref<ICommandHandler> callback = Ref<ICommandHandler>(), uint wakeupTimeMs = 0)
    {
        return OnAsync(cmd, pArg, callback, true, wakeupTimeMs);
    }
};
//--------------------------------------------------------- 
/**
 * @brief 基于CommandRunner的命令调度器
 * @date 2017-06-24 11:23
 *
 * 调度原理:
 * 1.将命令分为Except命令和功能命令
 * 2.Except命令为设置类命令,不涉及到设备
 * 3.功能命令不能冲突,每组命令之间(CommandSetter)通过一个锁进行异步
 * 4.运行时,以CommandSetter为一组进行调度
 */
class CommandAsyncWorker : public CommandRunner
{
public:
    typedef MapArguments<string, CommandAsyncDispatcher*>::EnumeratorType EnumeratorType;
protected:
    enum AsyncMode
    {
        OnAsyncMode = 0,
        TryAsyncMode,
        BackgroundMode
    };

    list_t<CommandAsyncDispatcher> _workObjItems;
    MapArguments<string, CommandAsyncDispatcher*> _workItems;
    MapArguments<CommandAsyncDispatcher*, CommandSetter> _workSetters;
    async_namespace::mutex _mutex;
    /// 分析Collection,配置异步命令分组
    void _AsyncCollection(const string& name, CommandCollection& collection)
    {
        list_t<CommandAsyncDispatcher>::iterator workItr = _workObjItems.push_back();
        CommandAsyncDispatcher* pDispatcher = &(*workItr);
        CommandSetter& setter = _workSetters.Put(pDispatcher);
        _workItems.Put(name, pDispatcher);

        CommandSetter exceptSetter;
        Ref<ICommandHandler> complexCmd;
        setter.Set(collection).All();
        exceptSetter.Set(collection);

        CommandInstaller::ExceptSysCommand(setter, &exceptSetter);
        CommandInstaller::ExceptCommand(setter, &exceptSetter);

        LOGGER(LoggerAdapter& _log = this->_log);
        LOGGER(_log << "Module:<" << name << ">\n";
        _log.WriteLine("Setter:");
        CommandSetter::EnumeratorType enumer = setter.GetEnumerator();
        while(enumer.MoveNext())
        {
            _log << enumer.Current()->Name() << " ";
        }
        _log.WriteLine();
        if(exceptSetter.Count() > 0)
        {
            _log.WriteLine("Except Setter:");
            enumer = exceptSetter.GetEnumerator();
            while(enumer.MoveNext())
            {
                _log << enumer.Current()->Name() << " ";
            }
        }
        _log.WriteLine());

        if(exceptSetter.Count() < 1)
            return;

        // Except 命令集
        workItr = _workObjItems.push_back();
        _workSetters.Put(&(*workItr)) = exceptSetter;

        string exceptName = name;
        exceptName += ".Except";
        _workItems.Put(exceptName, &(*workItr));
        return;
    }
    /// _logErr错误码
    bool _HandleErr(DeviceError::ErrorCode err, const char* sModule, const char* sCmd, const char* currentCmd = NULL)
    {
        if(DeviceError::IsSuccess(err))
            return true;
        string errMsg = CommandWorker::FormatErr(err, sModule, sCmd, currentCmd);
        ResetErr();
        _logErr(err, errMsg.c_str());
        return false;
    }
    /// 查找模块中指定的命令(包括Except集合)
    CommandAsyncDispatcher* _Find(const char* sModule, const char* sCmd, bool* pIsExcept = NULL)
    {
        string module = _strput(sModule);
        CommandAsyncDispatcher* pDispatcher = NULL;
        _workItems.Get(module, pDispatcher);
        if(pDispatcher == NULL)
            return NULL;
        CommandSetter& setter = _workSetters[pDispatcher].Value;
        // 如果查找不到,则继续在Except中查找
        if(setter.Seek(sCmd).IsNull())
        {
            pDispatcher = NULL;
            module += ".Except";
            _workItems.Get(module, pDispatcher);

            if(pIsExcept != NULL)
            {
                (*pIsExcept) = true;
            }
        }
        return pDispatcher;
    }
    /// 开始异步任务调度
    bool _Async(const char* sModule, const char* sCmd, const IStringArguments* pArg, Ref<ICommandHandler> callback, AsyncMode mode, uint wakeupTimeMs = 0)
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        string module = _strput(sModule);
        CommandAsyncDispatcher* pDispatcher = _Find(sModule, sCmd);
        DeviceError::ErrorCode err = DeviceError::UnKnownErr;
        if(pDispatcher != NULL)
        {
            CommandSetter& setter = _workSetters[pDispatcher].Value;
            Ref<ComplexCommand> complexCmd = setter.LookUp(sCmd);
            switch(mode)
            {
            case TryAsyncMode:
                err = pDispatcher->TryAsync(complexCmd, pArg, callback);
                break;
            case OnAsyncMode:
                err = pDispatcher->Async(complexCmd, pArg, callback);
                break;
            case BackgroundMode:
                err = pDispatcher->Background(complexCmd, pArg, callback, wakeupTimeMs);
                break;
            }
        }
        else
        {
            err = DeviceError::ArgRangeErr;
        }
        return _HandleErr(err, sModule, sCmd);
    }
public:
    CommandAsyncWorker() : CommandRunner() {}
    virtual ~CommandAsyncWorker()
    {
        Cancel();

        _workObjItems.clear();
        _workItems.Clear();
    }
    /// 获取调度器枚举器
    EnumeratorType GetEnumerator()
    {
        return _workItems.GetEnumerator();
    }
    /**
     * @brief 解析CommandRunner异步初始化
     * @date 2017-07-01 15:22
     */
    size_t OnWork()
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        LOGGER(LoggerAdapter& _log = this->_log);
        LOG_FUNC_NAME();
        _workItems.Clear();

        CommandRunner::MapEnumeratorType enumer = CommandRunner::_items.GetEnumerator();
        CommandCollection* pCollection = NULL;
        string name = "";
        size_t count = 1;
        _AsyncCollection(name, *this);
        while(enumer.MoveNext())
        {
            name = enumer.Current().Key;
            pCollection = enumer.Current().Value;
            _AsyncCollection(name, *pCollection);

            ++count;
        }
        return this->_logRetValue(count);
    }
    /// 返回是否已经异步初始化过
    inline bool IsAsync() const
    {
        return _workItems.Count() > 0;
    }
    /// 返回指定模块的任务队列执行器
    Ref<CommandAsyncDispatcher> Dispatcher(const char* sTask = NULL)
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        string module = _strput(sTask);
        CommandAsyncDispatcher* pDispatcher = NULL;
        _workItems.Get(module, pDispatcher);
        
        Ref<CommandAsyncDispatcher> dispatcher;
        if(pDispatcher != NULL)
        {
            dispatcher = (*pDispatcher);
        }
        return dispatcher;
    }
    /// 结束指定任务队列
    bool Cancel(const char* sTask, uint timeoutMs = 0)
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        string module = _strput(sTask);
        CommandAsyncDispatcher* pDispatcher = NULL;
        _workItems.Get(module, pDispatcher);

        if(pDispatcher != NULL)
        {
            pDispatcher->Cancel(timeoutMs);
            return true;
        }
        return false;
    }
    /// 结束所有任务
    void Cancel(uint timeoutMs = 0)
    {
        async_namespace::unique_lock<async_namespace::mutex> lock(_mutex);
        list_t<CommandAsyncDispatcher>::iterator itr;
        for(itr = _workObjItems.begin();itr != _workObjItems.end(); ++itr)
        {
            itr->Cancel(timeoutMs);
        }
    }
    /**
     * @brief 添加任务到异步队列中
     * @date 2017-07-01 15:24
     *
     * @param [in] sModule 需要操作的模块名
     * @param [in] sCmd 需要调用的命令
     * @param [in] pArg 调用时的参数
     * @param [in] callback 回调函数
     *
     * @return bool 返回调度的命令是否存在
     */
    inline bool Async(const char* sModule, const char* sCmd, const IStringArguments* pArg = NULL, 
        Ref<ICommandHandler> callback = Ref<ICommandHandler>())
    {
        return _Async(sModule, sCmd, pArg, callback, OnAsyncMode);
    }
    /**
     * @brief 尝试调用主队列中的命令
     * @date 2017-07-01 15:24
     *
     * @param [in] sModule 需要操作的模块名
     * @param [in] sCmd 需要调用的命令
     * @param [in] pArg 调用时的参数
     * @param [in] callback 回调函数
     *
     * @return bool 返回调度的命令是否成功调用(设备忙时返回false)
     */
    inline bool TryAsync(const char* sModule, const char* sCmd, const IStringArguments* pArg = NULL, 
        Ref<ICommandHandler> callback = Ref<ICommandHandler>())
    {
        return _Async(sModule, sCmd, pArg, callback, TryAsyncMode);
    }
    /**
     * @brief 添加任务到后台队列中
     * @date 2017-07-01 15:27
     *
     * @param [in] sModule 需要操作的模块名
     * @param [in] sCmd 需要调用的命令
     * @param [in] pArg 调用时的参数
     * @param [in] callback 回调函数
     * @param [in] wakeupTimeMs 调用的间隔时间
     *
     * @return bool 返回指定的命令是否存在
     */
    inline bool Background(const char* sModule, const char* sCmd, const IStringArguments* pArg = NULL, 
        Ref<ICommandHandler> callback =Ref<ICommandHandler>(), uint wakeupTimeMs = 0)
    {
        return _Async(sModule, sCmd, pArg, callback, BackgroundMode, wakeupTimeMs);
    }
    /**
     * @brief 等待所有任务完成
     * @date 2019-03-17 13:44:07
     */
    void Join()
    {
        MapArguments<string, CommandAsyncDispatcher*>::EnumeratorType enumer = _workItems.GetEnumerator();
        while(enumer.MoveNext())
        {
            enumer.Current().Value->Join();
        }
    }
    /**
     * @brief 等待队列中的前台任务完成
     * @date 2019-03-17 18:31:11
     */
    void Wait()
    {
        MapArguments<string, CommandAsyncDispatcher*>::EnumeratorType enumer = _workItems.GetEnumerator();
        while(enumer.MoveNext())
        {
            enumer.Current().Value->Wait();
        }
    }
};
//--------------------------------------------------------- 
} // namespace async
} // nemespace driver
} // namespace application
} // namespace zhou_yb
//--------------------------------------------------------- 
#endif // _LIBZHOUYB_COMMANDASYNCWORKER_H_
//========================================================= 